"""
General utility functions.
"""

import json
import logging
import sys
from pathlib import Path
from typing import Callable, Dict, List, Tuple, Union, Optional
from decimal import Decimal
from math import floor, ceil
from datetime import datetime, timedelta
import numpy as np
import talib
import pandas as pd
import threading
from .object import BarData, TickData
from .constant import Exchange, Interval


log_formatter = logging.Formatter('[%(asctime)s] %(message)s')


def extract_vt_symbol(vt_symbol: str) -> Tuple[str, Exchange]:
    """
    :return: (symbol, exchange)
    """
    symbol, exchange_str = vt_symbol.split(".")
    return symbol, Exchange(exchange_str)


def generate_vt_symbol(symbol: str, exchange: Exchange) -> str:
    """
    return vt_symbol
    """
    return f"{symbol}.{exchange.value}"


def _get_trader_dir(temp_name: str) -> Tuple[Path, Path]:
    """
    Get path where trader is running in.
    """
    cwd = Path.cwd()
    temp_path = cwd.joinpath(temp_name)

    # If .vntrader folder exists in current working directory,
    # then use it as trader running path.
    if temp_path.exists():
        return cwd, temp_path

    # Otherwise use home path of system.
    home_path = Path.home()
    temp_path = home_path.joinpath(temp_name)

    # Create .vntrader folder under home path if not exist.
    if not temp_path.exists():
        temp_path.mkdir()

    return home_path, temp_path


TRADER_DIR, TEMP_DIR = _get_trader_dir(".vntrader")
sys.path.append(str(TRADER_DIR))


def get_file_path(filename: str) -> Path:
    """
    Get path for temp file with filename.
    """
    return TEMP_DIR.joinpath(filename)


def get_folder_path(folder_name: str) -> Path:
    """
    Get path for temp folder with folder name.
    """
    folder_path = TEMP_DIR.joinpath(folder_name)
    if not folder_path.exists():
        folder_path.mkdir()
    return folder_path


def get_icon_path(filepath: str, ico_name: str) -> str:
    """
    Get path for icon file with ico name.
    """
    ui_path = Path(filepath).parent
    icon_path = ui_path.joinpath("ico", ico_name)
    return str(icon_path)


def load_json(filename: str) -> dict:
    """
    Load data from json file in temp path.
    """
    filepath = get_file_path(filename)

    if filepath.exists():
        with open(filepath, mode="r", encoding="UTF-8") as f:
            data = json.load(f)
        return data
    else:
        save_json(filename, {})
        return {}


def save_json(filename: str, data: dict) -> None:
    """
    Save data into json file in temp path.
    """
    filepath = get_file_path(filename)
    with open(filepath, mode="w+", encoding="UTF-8") as f:
        json.dump(
            data,
            f,
            indent=4,
            ensure_ascii=False
        )


def round_to(value: float, target: float) -> float:
    """
    Round price to price tick value.
    """
    value = Decimal(str(value))
    target = Decimal(str(target))
    rounded = float(int(round(value / target)) * target)
    return rounded


def floor_to(value: float, target: float) -> float:
    """
    Similar to math.floor function, but to target float number.
    """
    value = Decimal(str(value))
    target = Decimal(str(target))
    result = float(int(floor(value / target)) * target)
    return result


def ceil_to(value: float, target: float) -> float:
    """
    Similar to math.ceil function, but to target float number.
    """
    value = Decimal(str(value))
    target = Decimal(str(target))
    result = float(int(ceil(value / target)) * target)
    return result


def get_digits(value: float) -> int:
    """
    Get number of digits after decimal point.
    """
    value_str = str(value)

    if "e-" in value_str:
        _, buf = value_str.split("e-")
        return int(buf)
    elif "." in value_str:
        _, buf = value_str.split(".")
        return len(buf)
    else:
        return 0


# class BarGenerator:
#     """
#     For:
#     1. generating 1 minute bar data from tick data
#     2. generateing x minute bar/x hour bar data from 1 minute data

#     Notice:
#     1. for x minute bar, x must be able to divide 60: 2, 3, 5, 6, 10, 15, 20, 30
#     2. for x hour bar, x can be any number
#     """

#     def __init__(
#         self,
#         on_bar: Callable,
#         window: int = 0,
#         on_window_bar: Callable = None,
#         interval: Interval = Interval.MINUTE
#     ):
#         """Constructor"""
#         self.bar: BarData = None
#         self.on_bar: Callable = on_bar

#         self.interval: Interval = interval
#         self.interval_count: int = 0

#         self.hour_bar: BarData = None

#         self.window: int = window
#         self.window_bar: BarData = None
#         self.on_window_bar: Callable = on_window_bar

#         self.last_tick: TickData = None
#         self.last_bar: BarData = None

#     def update_tick(self, tick: TickData) -> None:
#         """
#         Update new tick data into generator.
#         """
#         new_minute = False

#         # Filter tick data with 0 last price
#         if not tick.last_price:
#             return

#         # Filter tick data with older timestamp
#         if self.last_tick and tick.datetime < self.last_tick.datetime:
#             return

#         if not self.bar:
#             new_minute = True
#         elif (
#             (self.bar.datetime.minute != tick.datetime.minute)
#             or (self.bar.datetime.hour != tick.datetime.hour)
#         ):
#             self.bar.datetime = self.bar.datetime.replace(
#                 second=0, microsecond=0
#             )
#             self.on_bar(self.bar)

#             new_minute = True

#         if new_minute:
#             self.bar = BarData(
#                 symbol=tick.symbol,
#                 exchange=tick.exchange,
#                 interval=Interval.MINUTE,
#                 datetime=tick.datetime,
#                 gateway_name=tick.gateway_name,
#                 open_price=tick.last_price,
#                 high_price=tick.last_price,
#                 low_price=tick.last_price,
#                 close_price=tick.last_price,
#                 open_interest=tick.open_interest
#             )
#         else:
#             self.bar.high_price = max(self.bar.high_price, tick.last_price)
#             if tick.high_price > self.last_tick.high_price:
#                 self.bar.high_price = max(self.bar.high_price, tick.high_price)

#             self.bar.low_price = min(self.bar.low_price, tick.last_price)
#             if tick.low_price < self.last_tick.low_price:
#                 self.bar.low_price = min(self.bar.low_price, tick.low_price)

#             self.bar.close_price = tick.last_price
#             self.bar.open_interest = tick.open_interest
#             self.bar.datetime = tick.datetime

#         if self.last_tick:
#             volume_change = tick.volume - self.last_tick.volume
#             self.bar.volume += max(volume_change, 0)

#         self.last_tick = tick

#     def update_bar(self, bar: BarData) -> None:
#         """
#         Update 1 minute bar into generator
#         """
#         if self.interval == Interval.MINUTE:
#             self.update_bar_minute_window(bar)
#         else:
#             self.update_bar_hour_window(bar)

#     def update_bar_minute_window(self, bar: BarData) -> None:
#         """"""
#         # If not inited, create window bar object
#         if not self.window_bar:
#             dt = bar.datetime.replace(second=0, microsecond=0)
#             self.window_bar = BarData(
#                 symbol=bar.symbol,
#                 exchange=bar.exchange,
#                 datetime=dt,
#                 gateway_name=bar.gateway_name,
#                 open_price=bar.open_price,
#                 high_price=bar.high_price,
#                 low_price=bar.low_price
#             )
#         # Otherwise, update high/low price into window bar
#         else:
#             self.window_bar.high_price = max(
#                 self.window_bar.high_price,
#                 bar.high_price
#             )
#             self.window_bar.low_price = min(
#                 self.window_bar.low_price,
#                 bar.low_price
#             )

#         # Update close price/volume into window bar
#         self.window_bar.close_price = bar.close_price
#         self.window_bar.volume += int(bar.volume)
#         self.window_bar.open_interest = bar.open_interest

#         # Check if window bar completed
#         if not (bar.datetime.minute + 1) % self.window:
#             self.on_window_bar(self.window_bar)
#             self.window_bar = None

#         # Cache last bar object
#         self.last_bar = bar

#     def update_bar_hour_window(self, bar: BarData) -> None:
#         """"""
#         # If not inited, create window bar object
#         if not self.hour_bar:
#             dt = bar.datetime.replace(minute=0, second=0, microsecond=0)
#             self.hour_bar = BarData(
#                 symbol=bar.symbol,
#                 exchange=bar.exchange,
#                 datetime=dt,
#                 gateway_name=bar.gateway_name,
#                 open_price=bar.open_price,
#                 high_price=bar.high_price,
#                 low_price=bar.low_price,
#                 volume=bar.volume
#             )
#             return

#         finished_bar = None

#         # If minute is 59, update minute bar into window bar and push
#         if bar.datetime.minute == 59:
#             self.hour_bar.high_price = max(
#                 self.hour_bar.high_price,
#                 bar.high_price
#             )
#             self.hour_bar.low_price = min(
#                 self.hour_bar.low_price,
#                 bar.low_price
#             )

#             self.hour_bar.close_price = bar.close_price
#             self.hour_bar.volume += int(bar.volume)
#             self.hour_bar.open_interest = bar.open_interest

#             finished_bar = self.hour_bar
#             self.hour_bar = None

#         # If minute bar of new hour, then push existing window bar
#         elif bar.datetime.hour != self.hour_bar.datetime.hour:
#             finished_bar = self.hour_bar

#             dt = bar.datetime.replace(minute=0, second=0, microsecond=0)
#             self.hour_bar = BarData(
#                 symbol=bar.symbol,
#                 exchange=bar.exchange,
#                 datetime=dt,
#                 gateway_name=bar.gateway_name,
#                 open_price=bar.open_price,
#                 high_price=bar.high_price,
#                 low_price=bar.low_price,
#                 volume=bar.volume
#             )
#         # Otherwise only update minute bar
#         else:
#             self.hour_bar.high_price = max(
#                 self.hour_bar.high_price,
#                 bar.high_price
#             )
#             self.hour_bar.low_price = min(
#                 self.hour_bar.low_price,
#                 bar.low_price
#             )

#             self.hour_bar.close_price = bar.close_price
#             self.hour_bar.volume += int(bar.volume)
#             self.hour_bar.open_interest = bar.open_interest

#         # Push finished window bar
#         if finished_bar:
#             self.on_hour_bar(finished_bar)

#         # Cache last bar object
#         self.last_bar = bar

#     def on_hour_bar(self, bar: BarData) -> None:
#         """"""
#         if self.window == 1:
#             self.on_window_bar(bar)
#         else:
#             if not self.window_bar:
#                 self.window_bar = BarData(
#                     symbol=bar.symbol,
#                     exchange=bar.exchange,
#                     datetime=bar.datetime,
#                     gateway_name=bar.gateway_name,
#                     open_price=bar.open_price,
#                     high_price=bar.high_price,
#                     low_price=bar.low_price
#                 )
#             else:
#                 self.window_bar.high_price = max(
#                     self.window_bar.high_price,
#                     bar.high_price
#                 )
#                 self.window_bar.low_price = min(
#                     self.window_bar.low_price,
#                     bar.low_price
#                 )

#             self.window_bar.close_price = bar.close_price
#             self.window_bar.volume += int(bar.volume)
#             self.window_bar.open_interest = bar.open_interest

#             self.interval_count += 1
#             if not self.interval_count % self.window:
#                 self.interval_count = 0
#                 self.on_window_bar(self.window_bar)
#                 self.window_bar = None

#     def generate(self) -> Optional[BarData]:
#         """
#         Generate the bar data and call callback immediately.
#         """
#         bar = self.bar

#         if self.bar:
#             bar.datetime = bar.datetime.replace(second=0, microsecond=0)
#             self.on_bar(bar)

#         self.bar = None
#         return bar
class BarGenerator(object):
    """
    K线合成器，支持：
    1. 基于Tick合成1分钟K线
    2. 基于1分钟K线合成X分钟K线（X可以是2、3、5、10、15、30、60）
    """
    # ----------------------------------------------------------------------
    def __init__(self, onBar, xmin=0, onXminBar=None, xSecond = 0, alignment='sharp',bias_second=2, marketClose = (23,59)):
        """Constructor"""
        self.bar = None  # 1分钟K线对象
        self.onBar = onBar  # 1分钟K线回调函数

        self.xminBar = None  # X分钟K线对象
        self.xmin = xmin  # X的值
        self.onXminBar = onXminBar  # X分钟K线的回调函数
        self.bias_second=bias_second
        self.hfBar = None  # 高频K线对象
        self.xSecond = xSecond
        self.onHFBar = onBar

        self.onCandle = onBar
        self.Candle = None
        self.onWCandle = onBar
        self.WeekCandle = None
        self.onMCandle = onBar
        self.MonthCandle = None

        self.lastTick = None  # 上一TICK缓存对象
        self.BarDone = None
        self.HfBarDone=None
        self.intraWeek = 0
        self.intraMonth = 0

        self.marketClose = marketClose
        self.alignment = alignment

    # ----------------------------------------------------------------------
    def update_align_tick(self, tick:TickData):
        """TICK更新"""
        # 初始化新一分钟的K线数据
        
        if self.lastTick and (tick.datetime-self.lastTick.datetime).total_seconds()>60:
            self.BarDone=None
            self.bar=None
            self.lastTick=None

        if self.bar and self.BarDone:
            
            if tick.datetime >= self.BarDone:
                self.bar.high_price = max(self.bar.high_price, tick.last_price)
                self.bar.low_price = min(self.bar.low_price, tick.last_price)
                self.bar.close_price = tick.last_price
                self.bar.open_interest = tick.open_interest
                
                # if tick.volumeChange:
                # self.bar.volume += tick.last_volume
                if self.lastTick:
                    self.bar.volume += (tick.volume - self.lastTick.volume)
                    ##money字段
                    self.bar.money+=(tick.volume - self.lastTick.volume)*tick.last_price  # 当前K线内的成交量（原版VNPY）

                self.lastTick = tick    # 缓存Tick
                # 推送已经结束的上一分钟K线
                self.bar.datetime=(self.BarDone+timedelta(seconds=self.bias_second))
                self.onBar(self.bar)
                # threading.Thread(target=self.onBar,args=(self.bar,)).start()
                self.bar = None
                return 

        
        
        if not self.bar:
            if self.BarDone and tick.datetime<=(self.BarDone+timedelta(seconds=self.bias_second)):
                self.lastTick = tick
                return 
            self.bar = BarData(
                gateway_name=tick.gateway_name,
                symbol=tick.symbol,
                exchange=tick.exchange,
                datetime=tick.datetime.replace(second=0, microsecond=0)
            )
            
            self.bar.symbol = tick.symbol
            self.bar.exchange = tick.exchange
       
            self.bar.open_price = tick.last_price
            self.bar.high_price = tick.last_price
            self.bar.low_price = tick.last_price
            
            
            self.bar.datetime = tick.datetime.replace(second=0, microsecond=0)  # 将秒和微秒设为0
            # self.bar.date = self.bar.datetime.strftime('%Y%m%d')
            # self.bar.time = self.bar.datetime.strftime('%H:%M:%S.%f')
            
            if self.BarDone:
                self.BarDone=self.BarDone+timedelta(minutes=1)
            else:
                #第一分钟只能拿到58s的数据，后面都是1min的数据
                self.BarDone = self.bar.datetime+timedelta(minutes=1)-timedelta(seconds=self.bias_second)

        self.bar.high_price = max(self.bar.high_price, tick.last_price)
        self.bar.low_price = min(self.bar.low_price, tick.last_price)
        self.bar.close_price = tick.last_price
        self.bar.open_interest = tick.open_interest
        
        # if tick.volumeChange:
        # self.bar.volume += tick.last_volume
        if self.lastTick:
            self.bar.volume += (tick.volume - self.lastTick.volume)
            ##money字段
            self.bar.money+=(tick.volume - self.lastTick.volume)*tick.last_price  # 当前K线内的成交量（原版VNPY）

        self.lastTick = tick    # 缓存Tick
   

    #---------------------------------------------------------------
    def update_sharp_tick(self, tick:TickData):
        """TICK更新"""
        
        if self.bar and self.BarDone:
            if tick.datetime >= self.BarDone:
                self.bar.high_price = max(self.bar.high_price, tick.last_price)
                self.bar.low_price = min(self.bar.low_price, tick.last_price)
                self.bar.close_price = tick.last_price
                self.bar.open_interest = tick.open_interest
                
                # if tick.volumeChange:
                # self.bar.volume += tick.last_volume
                if self.lastTick:
                    self.bar.volume += (tick.volume - self.lastTick.volume)
                    ##money字段
                    self.bar.money+=(tick.volume - self.lastTick.volume)*tick.last_price  # 当前K线内的成交量（原版VNPY）

                self.lastTick = tick    # 缓存Tick
                # 推送已经结束的上一分钟K线
                self.bar.datetime=(self.BarDone+timedelta(seconds=self.bias_second))
                # threading.Thread(target=self.onBar,args=(self.bar,)).start()
                self.onBar(self.bar)
                self.bar = None
                return 

        # 初始化新一分钟的K线数据
        if not self.bar:
            self.bar = BarData(
                gateway_name=tick.gateway_name,
                symbol=tick.symbol,
                exchange=tick.exchange,
                datetime=tick.datetime.replace(microsecond=0)
            )
            
            self.bar.symbol = tick.symbol
            self.bar.exchange = tick.exchange
       
            self.bar.open_price = tick.last_price
            self.bar.high_price = tick.last_price
            self.bar.low_price = tick.last_price
            
            
            self.bar.datetime = tick.datetime.replace(microsecond=0)  # 将秒和微秒设为0
            self.bar.date = self.bar.datetime.strftime('%Y%m%d')
            self.bar.time = self.bar.datetime.strftime('%H:%M:%S.%f')
            if self.lastTick and (tick.datetime-self.lastTick.datetime).total_seconds()>60:
                self.BarDone=None
            if self.BarDone:
                self.BarDone=self.BarDone+timedelta(minutes=1)
            else:
                #第一分钟只能拿到58s的数据，后面都是1min的数据
                self.BarDone = self.bar.datetime+timedelta(minutes=self.bias_second)-timedelta(seconds=2)

        self.bar.high_price = max(self.bar.high_price, tick.last_price)
        self.bar.low_price = min(self.bar.low_price, tick.last_price)
        self.bar.close_price = tick.last_price
        self.bar.open_interest = tick.open_interest
        
        # if tick.volumeChange:
        # self.bar.volume += tick.last_volume
        if self.lastTick:
            self.bar.volume += (tick.volume - self.lastTick.volume)
            ##money字段
            self.bar.money+=(tick.volume - self.lastTick.volume)*tick.last_price  # 当前K线内的成交量（原版VNPY）

        self.lastTick = tick    # 缓存Tick
    #---------------------------------------------------------------
    def update_hfBar(self,tick:TickData):
        
        # 高频交易的bar,高频的bar推送是提前0.5s，但是是收到下一个tick再发单，也就是收到tick的时间与bar时间一致，在高频bar里可以计算信号，不建议拿来发单，发单的话，在bar里或者tick里控制
        # tick.money=tick.volume*tick.last_price
        if self.HfBarDone and self.hfBar:
            if tick.datetime >= self.HfBarDone:
                # 累加更新老K线数据
                self.hfBar.high_price = max(self.hfBar.high_price, tick.last_price)
                self.hfBar.low_price = min(self.hfBar.low_price, tick.last_price)
                self.hfBar.close_price = tick.last_price
                self.hfBar.open_interest = tick.open_interest
                # self.hfBar.money+=tick.money
                # if tick.volumeChange:
                #     self.hfBar.volume += tick.last_volume
                if self.lastTick:
                    self.hfBar.volume += (tick.volume - self.lastTick.volume)
                    self.hfBar.money += (tick.volume - self.lastTick.volume)*tick.last_price
                    # 当前K线内的成交量（原版VNPY）
                # 
                self.lastTick = tick    # 缓存Tick
                # 推送已经结束的上一K线
                self.hfBar.datetime=(self.HfBarDone+timedelta(seconds=self.bias_second))
                
                # threading.Thread(target=self.onHFBar,args=(self.hfBar,)).start()
                self.onHFBar(self.hfBar)
                self.hfBar = None
                return 
                

        if not self.hfBar:
            
            self.hfBar = BarData(
                gateway_name=tick.gateway_name,
                symbol=tick.symbol,
                exchange=tick.exchange,
                datetime=tick.datetime.replace(microsecond=0)
            )
            self.hfBar.datetime = tick.datetime.replace(microsecond=0) 
            # 生成K线
            
            self.hfBar.symbol = tick.symbol
            self.hfBar.exchange = tick.exchange
            
             # 将微秒设为0
            self.hfBar.date = self.hfBar.datetime.strftime('%Y%m%d')
            self.hfBar.time = self.hfBar.datetime.strftime('%H:%M:%S.%f')
            if self.lastTick and ((tick.datetime-self.lastTick.datetime).total_seconds()>self.xSecond):
                self.HfBarDone=None
            if self.HfBarDone:
                self.HfBarDone=self.HfBarDone+timedelta(seconds=self.xSecond)
            else:
                #第一分钟只能拿到58s的数据，后面都是1min的数据
                self.HfBarDone = self.hfBar.datetime+timedelta(seconds=self.xSecond)-timedelta(seconds=self.bias_second)


            self.hfBar.open_price = tick.last_price
            self.hfBar.high_price = tick.last_price
            self.hfBar.low_price = tick.last_price

                            
        # 累加更新老K线数据
        self.hfBar.high_price = max(self.hfBar.high_price, tick.last_price)
        self.hfBar.low_price = min(self.hfBar.low_price, tick.last_price)
        self.hfBar.close_price = tick.last_price
        self.hfBar.open_interest = tick.open_interest
        # self.hfBar.money+=tick.money
        # if tick.volumeChange:
        #     self.hfBar.volume += tick.last_volume
        if self.lastTick:
            self.hfBar.volume += (tick.volume - self.lastTick.volume)
            self.hfBar.money += (tick.volume - self.lastTick.volume)*tick.last_price  # 当前K线内的成交量（原版VNPY）
        # 
        self.lastTick = tick    # 缓存Tick
            
    # ----------------------------------------------------------------------
    def update_bar(self, bar:BarData):
        """多分钟K线更新"""
        if self.alignment == 'full':
            # 尚未创建对象
            if not self.xminBar:
                self.xminBar = BarData(
                    gateway_name=bar.gateway_name,
                    symbol=bar.symbol,
                    exchange=bar.exchange,
                    datetime=bar.datetime
                )
                
                self.xminBar.symbol = bar.symbol
                self.xminBar.exchange = bar.exchange

                self.xminBar.open_price = bar.open_price
                self.xminBar.high_price = bar.high_price
                self.xminBar.low_price = bar.low_price
                # 生成上一X分钟K线的时间戳
                self.xminBar.datetime = bar.datetime.replace(second=0, microsecond=0)  # 将秒和微秒设为0
                self.xminBar.date = bar.datetime.strftime('%Y%m%d')
                self.xminBar.time = bar.datetime.strftime('%H:%M:%S.%f')
                self.BarDone = 0

            self.xminBar.high_price = max(self.xminBar.high_price, bar.high_price)
            self.xminBar.low_price = min(self.xminBar.low_price, bar.low_price)
            self.xminBar.close_price = bar.close_price
            self.xminBar.open_interest = bar.open_interest
            
            self.BarDone+=1

            if self.BarDone == self.xmin:
                self.onXminBar(self.xminBar)
                self.xminBar = None

        elif self.alignment == 'sharp':
            # X分钟已经走完
            if self.xminBar and self.BarDone:
                if  bar.datetime > self.BarDone:
                    self.onXminBar(self.xminBar)
                    # 清空老K线缓存对象
                    self.xminBar = None
            # 尚未创建对象
            if not self.xminBar:
                self.xminBar = BarData(
                    gateway_name=bar.gateway_name,
                    symbol=bar.symbol,
                    exchange=bar.exchange,
                    datetime=bar.datetime
                )
                
                self.xminBar.symbol = bar.symbol
                self.xminBar.exchange = bar.exchange

                self.xminBar.open_price = bar.open_price
                self.xminBar.high_price = bar.high_price
                self.xminBar.low_price = bar.low_price
                # 生成上一X分钟K线的时间戳
                self.xminBar.datetime = bar.datetime.replace(second=0, microsecond=0)  # 将秒和微秒设为0
                self.xminBar.date = bar.datetime.strftime('%Y%m%d')
                self.xminBar.time = bar.datetime.strftime('%H:%M:%S.%f')

                if self.xmin < 61:
                    diff = bar.datetime.minute % self.xmin
                    self.BarDone = self.xminBar.datetime + timedelta(seconds=(self.xmin-diff)*60-1)
                    
                elif self.xmin > 60:
                    diff = (bar.datetime.hour * 60 ) % self.xmin
                    self.BarDone = self.xminBar.datetime + timedelta(seconds=(self.xmin-diff)*60-1)

            self.xminBar.high_price = max(self.xminBar.high_price, bar.high_price)
            self.xminBar.low_price = min(self.xminBar.low_price, bar.low_price)
            self.xminBar.close_price= bar.close_price
            self.xminBar.open_interest = bar.open_interest
            self.xminBar.volume += bar.volume
            self.xminBar.money += bar.money
        
        if (bar.datetime.hour, bar.datetime.minute) == self.marketClose:   # 强制收盘切断
            self.onXminBar(self.xminBar)
            self.xminBar = None

    #----------------------------------------------------------------------
    def updateCandle(self, bar):
        """日K线更新"""
        # 尚未创建对象
        if not self.Candle:
            self.Candle = BarData(
                gateway_name=bar.gateway_name,
                symbol=bar.symbol,
                exchange=bar.exchange,
                datetime=bar.datetime
            )

            
            self.Candle.symbol = bar.symbol
            self.Candle.exchange = bar.exchange

            self.Candle.open_price = bar.open_price
            self.Candle.high_price = bar.high_price
            self.Candle.low_price = bar.low_price

        # 累加老K线
        self.Candle.datetime = bar.datetime.replace(second=0, microsecond=0)  # 将秒和微秒设为0
        self.Candle.date = bar.datetime.strftime('%Y%m%d')
        self.Candle.time = bar.datetime.strftime('%H:%M:%S.%f')
        self.Candle.high_price = max(self.Candle.high_price, bar.high_price)
        self.Candle.low_price = min(self.Candle.low_price, bar.low_price)
        self.Candle.close_price = bar.close_price
        self.Candle.open_interest = bar.open_interest
        self.Candle.volume += bar.volume

        # 推送
        if (bar.datetime.hour,bar.datetime.minute) == self.marketClose:  # 强制收盘切断
            if self.Candle:
                self.onCandle(self.Candle)
                # 清空老K线缓存对象
                self.Candle = None

    def updateWCandle(self, Candle):
        """周K线更新"""
        # 尚未创建对象
        abstract_week = Candle.datetime.strftime('%W')
        if not self.intraWeek:
            self.intraWeek = abstract_week
        if abstract_week != self.intraWeek:
            # 推送
            if self.WeekCandle:
                self.onWCandle(self.WeekCandle)
                # 清空老K线缓存对象
                self.WeekCandle = None

        if not self.WeekCandle:
            self.WeekCandle = BarData(
                gateway_name=Candle.gateway_name,
                symbol=Candle.symbol,
                exchange=Candle.exchange,
                datetime=Candle.datetime
            )

            
            self.WeekCandle.symbol = Candle.symbol
            self.WeekCandle.exchange = Candle.exchange

            self.WeekCandle.open_price = Candle.open_price
            self.WeekCandle.high_price = Candle.high_price
            self.WeekCandle.low_price = Candle.low_price
            self.WeekCandle.datetime = Candle.datetime.replace(second=0, microsecond=0)  # 将秒和微秒设为0
            self.WeekCandle.date = Candle.datetime.strftime('%Y%m%d')
            self.WeekCandle.time = Candle.datetime.strftime('%H:%M:%S.%f')

        # 累加老K线
        self.WeekCandle.high_price = max(self.WeekCandle.high_price, Candle.high_price)
        self.WeekCandle.low_price = min(self.WeekCandle.low_price, Candle.low_price)
        self.WeekCandle.close_price = Candle.close_price
        self.WeekCandle.open_interest = Candle.open_interest
        self.WeekCandle.volume += Candle.volume
        self.intraWeek = abstract_week

        if (self.xminBar.datetime.hour,self.xminBar.datetime.minute) == self.marketClose and self.marketClose != (23,59):
            if Candle.datetime.strftime('%w') == 5:  # 每周五收盘强切周线
                self.onWCandle(self.WeekCandle)
                self.WeekCandle = None
        elif (self.xminBar.datetime.hour,self.xminBar.datetime.minute) == self.marketClose and self.marketClose == (23,59):
            if Candle.datetime.strftime('%w') == 0:  # 7*24市场在周日晚0点切
                self.onWCandle(self.WeekCandle)
                self.WeekCandle = None


    #--------------------------            
    def generate(self):
        """手动强制立即完成K线合成"""
        self.onBar(self.bar)
        self.bar = None

# ########################################################################


class ArrayManager(object):
    """
    For:
    1. time series container of bar data
    2. calculating technical indicator value
    """

    def __init__(self, size: int = 100,mode="normal"):
        """Constructor"""
        self.mode=mode
        if self.mode=="normal":
            self.count: int = 0
            self.size: int = size
            self.inited: bool = False

            self.open_array: np.ndarray = np.zeros(size)
            self.high_array: np.ndarray = np.zeros(size)
            self.low_array: np.ndarray = np.zeros(size)
            self.close_array: np.ndarray = np.zeros(size)
            self.volume_array: np.ndarray = np.zeros(size)
            self.open_interest_array: np.ndarray = np.zeros(size)
            self.money_array:np.ndarray=np.zeros(size)

            self.datetime_array:np.ndarray=np.array([datetime.now()]*size)
        if self.mode=="min":
            self.count:int =0
            self.size:int = size
            self.inited:bool=False
            self.open_list:List[float]=[]
            self.high_list: List[float] = []
            self.low_list: List[float] = []
            self.close_list: List[float] = []
            self.volume_list: List[float] = []
            self.open_interest_list: List[float] = []
            self.money_list:List[float]=[]
            self.datetime_list:List[datetime]=[]

    def update_bar(self, bar: BarData) -> None:
        """
        Update new bar data into array manager.
        """
        if self.mode=="normal":
            self.count += 1
            if not self.inited and self.count >= self.size:
                self.inited = True

            self.open_array[:-1] = self.open_array[1:]
            self.high_array[:-1] = self.high_array[1:]
            self.low_array[:-1] = self.low_array[1:]
            self.close_array[:-1] = self.close_array[1:]
            self.volume_array[:-1] = self.volume_array[1:]
            self.open_interest_array[:-1] = self.open_interest_array[1:]
            self.money_array[:-1]=self.money_array[1:]
            self.datetime_array[:-1]=self.datetime_array[1:]

            self.open_array[-1] = bar.open_price
            self.high_array[-1] = bar.high_price
            self.low_array[-1] = bar.low_price
            self.close_array[-1] = bar.close_price
            self.volume_array[-1] = bar.volume
            self.open_interest_array[-1] = bar.open_interest
            self.money_array[-1]=bar.money
            self.datetime_array[-1]=bar.datetime
        if self.mode=="min":
            self.count += 1
            if not self.inited and self.count >= self.size:
                self.inited = True
            self.open_list.append(bar.open_price)
            self.high_list.append(bar.high_price)
            self.low_list.append(bar.low_price)
            self.close_list.append(bar.close_price)
            self.volume_list.append(bar.volume)
            self.open_interest_list.append(bar.open_interest)
            self.money_list.append(bar.money)
            self.datetime_list.append(bar.datetime)

    @property
    def open(self):
        """
        Get open price time series.
        """
        if self.mode=="normal":
            return self.open_array
        if self.mode=="min":
            return self.open_list

    @property
    def high(self):
        """
        Get high price time series.
        """
        if self.mode=="normal":
            return self.high_array
        if self.mode=="min":
            return self.high_list

    @property
    def low(self) :
        """
        Get low price time series.
        """
        if self.mode=="normal":
            return self.low_array
        if self.mode=="min":
            return self.low_list

    @property
    def close(self):
        """
        Get close price time series.
        """
        if self.mode=="normal":
            return self.close_array
        if self.mode=="min":
            return self.close_list

    @property
    def volume(self):
        """
        Get trading volume time series.
        """
        if self.mode=="normal":
            return self.volume_array
        if self.mode=="min":
            return self.volume_list

    @property
    def open_interest(self):
        """
        Get trading volume time series.
        """
        if self.mode=="normal":
            return self.open_interest_array
        if self.mode=="min":
            return self.open_interest_list

    @property
    def money(self):
        """
        Get trading money time series.
        """
        if self.mode=="normal":
            return self.money_array
        if self.mode=="min":
            return self.money_list

    @property
    def datetime(self):
        """
        Get trading datetime time series.
        """
        if self.mode=="normal":
            return self.datetime_array
        if self.mode=="min":
            return self.datetime_list
    
    def ohlcvom(self)->np.ndarray:
        return np.array([self.open,self.high,self.low,self.close,self.volume,self.open_interest,self.money]).T
    
    def sma(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        Simple moving average.
        """
        result = talib.SMA(self.close, n)
        if array:
            return result
        return result[-1]

    def ema(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        Exponential moving average.
        """
        result = talib.EMA(self.close, n)
        if array:
            return result
        return result[-1]

    def kama(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        KAMA.
        """
        result = talib.KAMA(self.close, n)
        if array:
            return result
        return result[-1]

    def wma(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        WMA.
        """
        result = talib.WMA(self.close, n)
        if array:
            return result
        return result[-1]

    def apo(
        self,
        fast_period: int,
        slow_period: int,
        matype: int = 0,
        array: bool = False
    ) -> Union[float, np.ndarray]:
        """
        APO.
        """
        result = talib.APO(self.close, fast_period, slow_period, matype)
        if array:
            return result
        return result[-1]

    def cmo(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        CMO.
        """
        result = talib.CMO(self.close, n)
        if array:
            return result
        return result[-1]

    def mom(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        MOM.
        """
        result = talib.MOM(self.close, n)
        if array:
            return result
        return result[-1]

    def ppo(
        self,
        fast_period: int,
        slow_period: int,
        matype: int = 0,
        array: bool = False
    ) -> Union[float, np.ndarray]:
        """
        PPO.
        """
        result = talib.PPO(self.close, fast_period, slow_period, matype)
        if array:
            return result
        return result[-1]

    def roc(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        ROC.
        """
        result = talib.ROC(self.close, n)
        if array:
            return result
        return result[-1]

    def rocr(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        ROCR.
        """
        result = talib.ROCR(self.close, n)
        if array:
            return result
        return result[-1]

    def rocp(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        ROCP.
        """
        result = talib.ROCP(self.close, n)
        if array:
            return result
        return result[-1]

    def rocr_100(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        ROCR100.
        """
        result = talib.ROCR100(self.close, n)
        if array:
            return result
        return result[-1]

    def trix(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        TRIX.
        """
        result = talib.TRIX(self.close, n)
        if array:
            return result
        return result[-1]

    def std(self, n: int, nbdev: int = 1, array: bool = False) -> Union[float, np.ndarray]:
        """
        Standard deviation.
        """
        result = talib.STDDEV(self.close, n, nbdev)
        if array:
            return result
        return result[-1]

    def obv(self, array: bool = False) -> Union[float, np.ndarray]:
        """
        OBV.
        """
        result = talib.OBV(self.close, self.volume)
        if array:
            return result
        return result[-1]

    def cci(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        Commodity Channel Index (CCI).
        """
        result = talib.CCI(self.high, self.low, self.close, n)
        if array:
            return result
        return result[-1]

    def atr(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        Average True Range (ATR).
        """
        result = talib.ATR(self.high, self.low, self.close, n)
        if array:
            return result
        return result[-1]

    def natr(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        NATR.
        """
        result = talib.NATR(self.high, self.low, self.close, n)
        if array:
            return result
        return result[-1]

    def rsi(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        Relative Strenght Index (RSI).
        """
        result = talib.RSI(self.close, n)
        if array:
            return result
        return result[-1]

    def macd(
        self,
        fast_period: int,
        slow_period: int,
        signal_period: int,
        array: bool = False
    ) -> Union[
        Tuple[np.ndarray, np.ndarray, np.ndarray],
        Tuple[float, float, float]
    ]:
        """
        MACD.
        """
        macd, signal, hist = talib.MACD(
            self.close, fast_period, slow_period, signal_period
        )
        if array:
            return macd, signal, hist
        return macd[-1], signal[-1], hist[-1]

    def adx(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        ADX.
        """
        result = talib.ADX(self.high, self.low, self.close, n)
        if array:
            return result
        return result[-1]

    def adxr(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        ADXR.
        """
        result = talib.ADXR(self.high, self.low, self.close, n)
        if array:
            return result
        return result[-1]

    def dx(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        DX.
        """
        result = talib.DX(self.high, self.low, self.close, n)
        if array:
            return result
        return result[-1]

    def minus_di(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        MINUS_DI.
        """
        result = talib.MINUS_DI(self.high, self.low, self.close, n)
        if array:
            return result
        return result[-1]

    def plus_di(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        PLUS_DI.
        """
        result = talib.PLUS_DI(self.high, self.low, self.close, n)
        if array:
            return result
        return result[-1]

    def willr(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        WILLR.
        """
        result = talib.WILLR(self.high, self.low, self.close, n)
        if array:
            return result
        return result[-1]

    def ultosc(
        self,
        time_period1: int = 7,
        time_period2: int = 14,
        time_period3: int = 28,
        array: bool = False
    ) -> Union[float, np.ndarray]:
        """
        Ultimate Oscillator.
        """
        result = talib.ULTOSC(self.high, self.low, self.close, time_period1, time_period2, time_period3)
        if array:
            return result
        return result[-1]

    def trange(self, array: bool = False) -> Union[float, np.ndarray]:
        """
        TRANGE.
        """
        result = talib.TRANGE(self.high, self.low, self.close)
        if array:
            return result
        return result[-1]

    def boll(
        self,
        n: int,
        dev: float,
        array: bool = False
    ) -> Union[
        Tuple[np.ndarray, np.ndarray],
        Tuple[float, float]
    ]:
        """
        Bollinger Channel.
        """
        mid = self.sma(n, array)
        std = self.std(n, 1, array)

        up = mid + std * dev
        down = mid - std * dev

        return up, down

    def keltner(
        self,
        n: int,
        dev: float,
        array: bool = False
    ) -> Union[
        Tuple[np.ndarray, np.ndarray],
        Tuple[float, float]
    ]:
        """
        Keltner Channel.
        """
        mid = self.sma(n, array)
        atr = self.atr(n, array)

        up = mid + atr * dev
        down = mid - atr * dev

        return up, down

    def donchian(
        self, n: int, array: bool = False
    ) -> Union[
        Tuple[np.ndarray, np.ndarray],
        Tuple[float, float]
    ]:
        """
        Donchian Channel.
        """
        up = talib.MAX(self.high, n)
        down = talib.MIN(self.low, n)

        if array:
            return up, down
        return up[-1], down[-1]

    def aroon(
        self,
        n: int,
        array: bool = False
    ) -> Union[
        Tuple[np.ndarray, np.ndarray],
        Tuple[float, float]
    ]:
        """
        Aroon indicator.
        """
        aroon_up, aroon_down = talib.AROON(self.high, self.low, n)

        if array:
            return aroon_up, aroon_down
        return aroon_up[-1], aroon_down[-1]

    def aroonosc(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        Aroon Oscillator.
        """
        result = talib.AROONOSC(self.high, self.low, n)

        if array:
            return result
        return result[-1]

    def minus_dm(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        MINUS_DM.
        """
        result = talib.MINUS_DM(self.high, self.low, n)

        if array:
            return result
        return result[-1]

    def plus_dm(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        PLUS_DM.
        """
        result = talib.PLUS_DM(self.high, self.low, n)

        if array:
            return result
        return result[-1]

    def mfi(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
        """
        Money Flow Index.
        """
        result = talib.MFI(self.high, self.low, self.close, self.volume, n)
        if array:
            return result
        return result[-1]

    def ad(self, array: bool = False) -> Union[float, np.ndarray]:
        """
        AD.
        """
        result = talib.AD(self.high, self.low, self.close, self.volume)
        if array:
            return result
        return result[-1]

    def adosc(
        self,
        fast_period: int,
        slow_period: int,
        array: bool = False
    ) -> Union[float, np.ndarray]:
        """
        ADOSC.
        """
        result = talib.ADOSC(self.high, self.low, self.close, self.volume, fast_period, slow_period)
        if array:
            return result
        return result[-1]

    def bop(self, array: bool = False) -> Union[float, np.ndarray]:
        """
        BOP.
        """
        result = talib.BOP(self.open, self.high, self.low, self.close)

        if array:
            return result
        return result[-1]


def virtual(func: Callable) -> Callable:
    """
    mark a function as "virtual", which means that this function can be override.
    any base class should use this or @abstractmethod to decorate all functions
    that can be (re)implemented by subclasses.
    """
    return func


file_handlers: Dict[str, logging.FileHandler] = {}


def _get_file_logger_handler(filename: str) -> logging.FileHandler:
    handler = file_handlers.get(filename, None)
    if handler is None:
        handler = logging.FileHandler(filename)
        file_handlers[filename] = handler  # Am i need a lock?
    return handler


def get_file_logger(filename: str) -> logging.Logger:
    """
    return a logger that writes records into a file.
    """
    logger = logging.getLogger(filename)
    handler = _get_file_logger_handler(filename)  # get singleton handler.
    handler.setFormatter(log_formatter)
    logger.addHandler(handler)  # each handler will be added only once.
    return logger

